package com.markhsiu.MiniMQ.customer;

import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;

import com.markhsiu.MiniMQ.core.constant.ConstantUtil;
import com.markhsiu.MiniMQ.model.Message;
import com.markhsiu.MiniMQ.model.RequestMessage;
import com.markhsiu.MiniMQ.model.Result;
import com.markhsiu.MiniMQ.model.constant.MessageSourceEnum;
import com.markhsiu.MiniMQ.remote.Client;
import com.markhsiu.MiniMQ.remote.transport.netty.NettyClient;

/**
 * Created by Mark Hsiu on 2017/2/8.
 */
public class Customer {

	private static final String body = "customer1 send";
	private Client client = new NettyClient();
	private final Queue<Result> queue = new ConcurrentLinkedQueue<Result>();

	public Customer() {}

	public Customer(Client client) {
		this.client = client;
	}

	public void start(final String topic) {
		client.open();
		client.connect();
		
		ConstantUtil.threadPools(new Runnable() {
			
			@Override
			public void run() {
				Customer.this.run(topic);
			}
		});
	}

 
	private void run(final String topic){
		while (true) {
			Message message = new RequestMessage();
			message.setMessageID(UUID.randomUUID().toString());
			message.setTopic(topic);
			message.setBody(body.getBytes());
			message.setTimestamp(System.nanoTime());
			message.setSource(MessageSourceEnum.CUSTOMER);
			message.setTarget(MessageSourceEnum.BROKER);
			Result result = client.send(message);
			if (result.getError() == 0 || result.getMessage().getBody() != null) {
				String bo = new String(result.getMessage().getBody());
				if (bo.equals("ONWAY")) {
					try {
						TimeUnit.SECONDS.sleep(2);
						continue;	
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
				queue.add(result);
			}
			
		}
	}
	
	public Result getMessage(){
		return queue.poll();
	}

	public void end() {
		client.close();
	}
	
}
